package cometd.test;

import java.io.IOException;
import java.io.PrintWriter;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.catalina.CometEvent;
import org.apache.catalina.CometProcessor;

public class ChatServlet extends HttpServlet implements CometProcessor 
{

	protected static Map<String,Collection<Connection>> connections = new HashMap<String,Collection<Connection>>();
	protected static Map<String,Date> lastUpdate = new HashMap<String,Date>();
	protected static NotifierThread notifierThread = null;

	protected static Long nextId=0l;
	
	@Override
	public void init() throws ServletException {
		notifierThread = new NotifierThread();
		Thread messageSenderThread = new Thread(notifierThread, "MessageSender["+ getServletContext().getContextPath() + "]");
		messageSenderThread.setDaemon(true);
		messageSenderThread.start();
	}
	
	@Override
	public void destroy() 
	{
		connections.clear();
		notifierThread.stop();
		notifierThread = null;
	}

	public static long getNextId()
	{
		synchronized (nextId) 
		{
			nextId = nextId+1;
			return nextId;
		}
	}
	
	/**
	 * Process the given Comet event.
	 * 
	 * @param event
	 *            The Comet event that will be processed
	 * @throws IOException
	 * @throws ServletException
	 */
	public void event(CometEvent event) throws IOException, ServletException {
		HttpServletRequest request = event.getHttpServletRequest();
		HttpServletResponse response = event.getHttpServletResponse();
		//System.out.println("Event:" + event.getEventType() );
		if (event.getEventType() == CometEvent.EventType.BEGIN) {
			synchronized (connections) {
				addConnection(request, response);
			}
		} else if (event.getEventType() == CometEvent.EventType.ERROR) {
			synchronized (connections) {
				removeConnection(request, response);
			}
			event.close();
		} else if (event.getEventType() == CometEvent.EventType.END) {
			synchronized (connections) {
				removeConnection(request, response);
			}
			event.close();
		} else if (event.getEventType() == CometEvent.EventType.READ) {
			// ignore input from stream
		}
	}

	private class NotifierThread implements Runnable 
	{

		protected boolean running = true;
		protected Collection<Notification> notifications = new HashSet<Notification>();

		public NotifierThread() {
		}

		public void stop() 
		{
			running = false;
		}

		/**
		 * Add message for sending.
		 */
		public void notifyConnections( Long connectionId, String id ) 
		{
			synchronized (notifications) {
				notifications.add( new Notification( connectionId, id ) );
				notifications.notify();
			}
		}

		public void run() 
		{

			while (running) {
				try {
					if (notifications.size() == 0) 
					{
						synchronized (notifications) 
						{
							notifications.wait();
						}
					}
					synchronized (connections) 
					{
						Notification[] pendingNotifications = null;
						synchronized (notifications) {
							pendingNotifications = notifications.toArray(new Notification[0]);
							notifications.clear();
						}
						// Send any pending message on all the open connections
						for( Notification notification : pendingNotifications )
						{
							Date d = new Date(); 
							lastUpdate.put( notification.channel, d );
							Collection<Connection> channelConnections = connections.get( notification.channel );
							if( channelConnections!=null )
							{
								List<Connection> remove = new LinkedList<Connection>();
								for (Connection connection: channelConnections ) 
								{
									try{
										PrintWriter writer = connection.response.getWriter();
										if( notification.fromConnectionId != connection.connectionId )
											writer.write( "onNotify( '"+notification.channel+"', "+d.getTime()+");" );
										else
											writer.write( "lastUpdate="+d.getTime()+";" );
										writer.flush();
										writer.close();
									}catch( Exception e )
									{
										remove.add( connection );
									}
								}
								if( !remove.isEmpty() )channelConnections.removeAll(remove);
								if( channelConnections.isEmpty() )
									connections.remove( notification.channel );
							}
						}
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
		
	}
	
	private void addConnection( HttpServletRequest request, HttpServletResponse response )
	{
		Date lu = new Date( Long.parseLong( request.getParameter("lastUpdate") ) );
		System.out.println( "new connection Listening to:"+request.getParameter("channels")+" "+lu );
		if( lu.getTime()!=0 ) // accept connection if the client is in its inital state.
		{
			Date latest=null;
			String allChannelsThatHaveBeenNotified = "";
			for( String channel : request.getParameter("channels").split("_") )
			{
				Date current = lastUpdate.get( channel );
				if( current!=null )
				{
					System.out.println( "last update for "+channel+" has been at:"+current );
					allChannelsThatHaveBeenNotified += (allChannelsThatHaveBeenNotified.length()!=0?"_":"")+channel;
					if( (latest==null || latest.before(current))   )
						latest = current;
				}
			}
	
			if( latest!=null && lu.before(latest) ) 
			{
				System.out.println( "directly notifying" );
				try{
					PrintWriter writer = response.getWriter();
					writer.write( "onNotify( '"+allChannelsThatHaveBeenNotified+"', "+latest.getTime()+");" );
					writer.flush();
					writer.close();
					return;
				}catch(Exception e)
				{}
			}
		}

		
		for( String channel : request.getParameter("channels").split("_") )
		{
			Collection<Connection> responses = connections.get( channel );
			if( responses==null )
			{
				responses = new LinkedList<Connection>();
				connections.put( channel, responses );
			}
			responses.add( new Connection( request, response ) );
		}
	}
	
	private void removeConnection( HttpServletRequest request, HttpServletResponse response)
	{
		if( request.getParameter("channels")!=null ) 
		{
			for( String channel : request.getParameter("channels").split("_") )
			{
				Collection<Connection> responses = connections.get( channel );
				if( responses!=null )
				{
					responses.remove( new Connection( request, response ) );
					if( responses.isEmpty() )
						connections.remove( channel );
				}
			}
		}
	}
	
	public static void notify( long connectionId, String id ) 
	{
		notifierThread.notifyConnections( connectionId, id );
	}

	private class Connection
	{
		
		public Connection(HttpServletRequest request, HttpServletResponse response) 
		{
			connectionId = Long.parseLong( request.getParameter( "connectionId" ) );
			this.response = response;
		}
		long connectionId;
		HttpServletResponse response;
		
		@Override
		public boolean equals(Object obj) 
		{
			if( !(obj instanceof Connection) )return false;
			Connection c =(Connection)obj;
			return c.connectionId==connectionId && c.response.equals(response);
		}
	}

	private class Notification
	{
		long fromConnectionId;
		String channel;
		
		public Notification(long fromConnectionId, String channel) 
		{
			this.fromConnectionId = fromConnectionId;
			this.channel = channel;
		}
	}
	
}
